package com.shujia.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

public class HbaseTool {
    public static final Connection CONNECTION;

    public static final Admin ADMIN;

    private HbaseTool() {
    }

    //静态代码块
    static {
        Connection tmpConn = null;
        Admin tmpAdmin = null;
        try {
            //读取配置文件，获取zookeeper信息
            Properties properties = new Properties();
            properties.load(new BufferedReader(new FileReader("hbase/src/main/resources/infos.properties")));

            String zkInfo = properties.getProperty("hbase.zookeeper.quorum");

            Configuration conf = HBaseConfiguration.create();
            //配置zookeeper集群信息
            conf.set("hbase.zookeeper.quorum", zkInfo);

            //获取数据库连接对象
            tmpConn = ConnectionFactory.createConnection(conf);

            //获取数据库操作对象
            tmpAdmin = tmpConn.getAdmin();
        } catch (Exception e) {
            e.printStackTrace();
        }
        CONNECTION = tmpConn;
        ADMIN = tmpAdmin;
    }

    //获取表对象
    public static Table getOneTable(String tableName) {
        Table t = null;
        try {
            //先将表名封装成TableName对象
            TableName tn = TableName.valueOf(tableName);

            //先判断是否存在
            boolean b = ADMIN.tableExists(tn);
            if (b) {
                //获取表对象
                t = CONNECTION.getTable(tn);
            } else {
                System.out.println(Bytes.toString(tn.getName()) + "表不存在！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return t;
    }

    //创建hbase表
    public static void createHbaseTable(String tableName, String... cfs) {
        try {
            TableName tn = TableName.valueOf(tableName);
            //应该在创建表之前先判断表是否存在
            boolean b = ADMIN.tableExists(tn);
            if (!b) {
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tn);

                //创建一个存储列簇描述器的集合
                ArrayList<ColumnFamilyDescriptor> cfList = new ArrayList<>();

                for (String cf : cfs) {
                    //新版本创建列簇描述器并加入到表描述器中的做法
//                    ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder.of(cf); // ROW
                    //创建列簇同时设置布隆过滤器
                    ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder
                            .newBuilder(Bytes.toBytes(cf))
                            .setBloomFilterType(BloomType.ROWCOL)
                            .build();

                    String s = cfd.getBloomFilterType().toString();
                    System.out.println("------------------");
                    System.out.println(s);
                    System.out.println("------------------");
//                    HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);
//                    hColumnDescriptor.setBloomFilterType(BloomType.ROW);

                    cfList.add(cfd);
                }

                tableDescriptorBuilder.setColumnFamilies(cfList);

                //void createTable(TableDescriptor desc)
                ADMIN.createTable(tableDescriptorBuilder.build());
                System.out.println(Bytes.toString(tn.getName()) + " 表创建成功！");
            } else {
                System.out.println(Bytes.toString(tn.getName()) + "表已经存在！！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //删除表
    public static void deleteTable(String tName) {
        try {
            //先将表名封装成TableName对象
            TableName tableName = TableName.valueOf(tName);
            //先判断是否存在
            boolean b = ADMIN.tableExists(tableName);
            if (b) {
                ADMIN.disableTable(tableName);
                ADMIN.deleteTable(tableName);
            } else {
                System.out.println(Bytes.toString(tableName.getName()) + "表不存在！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //查询一批数据
    public static Iterator<Result> scanData(String tn) {
        Iterator<Result> resultIterator = null;
        try {
            //先将表名封装成TableName对象
            TableName tableName = TableName.valueOf(tn);

            //先判断是否存在
            boolean b = ADMIN.tableExists(tableName);
            if (b) {
                //获取表对象
                Table students = CONNECTION.getTable(tableName);

                Scan scan = new Scan(); //若使用无参构造方法创建Scan对象，表示全表扫描
                //ResultScanner getScanner(Scan scan)
                ResultScanner rs = students.getScanner(scan); // rs中存储了多行数据
                resultIterator = rs.iterator(); //获取所有行数据组成的迭代器

            } else {
                System.out.println(Bytes.toString(tableName.getName()) + "表不存在！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return resultIterator;
    }


    //TODO:自己编写
//    public static Iterator<Result> scanData(String tableName, int limit){
//
//    }

    //创建预分region表 "test3", "cf1","cf2","f","m"
    public static void createPreviewTable(String tn, List<String> cfs, List<String> splitRowKeys) {
        try {
            //先将表名封装成TableName对象
            TableName tableName = TableName.valueOf(tn);

            //创建表描述器对象
            TableDescriptorBuilder tdb = TableDescriptorBuilder.newBuilder(tableName);

            //创建列簇描述器对象
            //创建一个存储列簇描述器的集合
            ArrayList<ColumnFamilyDescriptor> cfList = new ArrayList<>();
            for (String cf : cfs) {
                //新版本创建列簇描述器并加入到表描述器中的做法
                ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder.of(cf);
                cfList.add(cfd);
            }
            //将列簇加入到表中
            tdb.setColumnFamilies(cfList);

            List<byte[]> splitList = splitRowKeys
                    .stream()
                    .map(Bytes::toBytes).collect(Collectors.toList()); // List<"f","m">

            byte[][] splitKeys = new byte[splitList.size()][];

            for (int i = 0; i < splitList.size(); i++) {
                splitKeys[i] = splitList.get(i);
            }

            //创建预分region表
            //createTable(TableDescriptor desc, byte[][] splitKeys)
            ADMIN.createTable(tdb.build(), splitKeys); // 切分出的region由master负载均衡，会有多个regionserver所管理

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //遍历Result结果的方法
    public static void printResult(Result result) {
        // 获取一行所有的列单元格
        List<Cell> cells = result.listCells();
        //获取行键
        String rowKey = Bytes.toString(result.getRow());
        StringBuilder sb = new StringBuilder();
        sb.append("行键：").append(rowKey).append(", ");
        for (Cell cell : cells) {
            String colName = Bytes.toString(CellUtil.cloneQualifier(cell));
            String colValue = Bytes.toString(CellUtil.cloneValue(cell));
            sb.append(colName).append("：").append(colValue).append(", ");
        }
        String str = sb.toString();
        str = str.substring(0, str.length() - 2);
        System.out.println(str);
    }

    public static void close() {
        if (ADMIN != null) {
            try {
                ADMIN.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if (CONNECTION != null) {
            try {
                CONNECTION.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
